package com.atguigu.spark.util

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.UUID
import java.util.concurrent.TimeUnit

object Spark_Close {
  def main(args: Array[String]): Unit = {

    // TODO Spark Streaming Env
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount").set("spark.testing.memory", "2147480000")
    val ssc = new StreamingContext(conf, Seconds(3))

    val dstream: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver)

    dstream.print()

    ssc.start()
    new Thread(()=>{
      Thread.sleep(5)
      ssc.stop(stopSparkContext = true,stopGracefully = true)
    }).start()
    ssc.awaitTermination()
  }
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY){
    private var flg = true
    override def onStart(): Unit = {

      // 采集数据
      while ( flg ) {

        // 生成数据
        val uuid: String = UUID.randomUUID().toString
        // 存储数据
        this.store(uuid)
        Thread.sleep(1000)

      }
    }

    override def onStop(): Unit = {
      flg = false
    }
  }
}
